package com.neusoft.bd.myflink.step06;

import com.neusoft.bd.myflink.entity.WaterSensor;
import com.neusoft.bd.myflink.source.WaterSensorSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichAggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

public class WindowAggregateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> stream =
                env.fromSource(
                        WaterSensorSource.waterSensorDataStreamSource(),
                        WatermarkStrategy.noWatermarks(),
                        "Generator Source");
        WindowedStream<WaterSensor, String, TimeWindow> window = stream.keyBy(ws -> ws.getId())
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)));
        window.aggregate(new RichAggregateFunction<WaterSensor, Integer, String>() {
            @Override
            public Integer createAccumulator() {


                return 0;
            }

            @Override
            public Integer add(WaterSensor ws, Integer accumulator) {

                return ws.getVc() + accumulator;
            }

            @Override
            public String getResult(Integer accumulator) {
                return "Result --  > " + accumulator;
            }

            @Override
            public Integer merge(Integer a, Integer b) {
                return null;
            }
        }).print();

        env.execute();


    }
}
